Building a Crypto Backtester with the Coinbase API — Part 2: Code setup

This is the follow-up article from the first part. If you haven’t already read that article, please do and come back.

Requirements for the guide:

• You need to be able understand wrapper functions in Python

Creating all the classes

We’ll first make every folder a python module. To do this, in each folder create an empty file named __init__.py

Rest client wrapper

Then in the clients folder, add a file named rest_client.py . In the file, add an empty class definition for our rest client:

class RESTClientWrapper:
    def __init__(self):
        pass

We call this a wrapper, since it encloses the rest client provided by coinbase and makes it easier for us to use it. We’d need it to load in our environment variables and input them into the actual rest client.

Rest client poller

In the data_collection folder, add a file named rest_client_poller.py . In this file add an empty class definition for our poller. We will use this to run the actual market queries on an interval and store the data.

class RESTClientPoller:
    def __init__(self):
        pass

Abstract indicator class

In the data_processing folder, create a file named indicator.py . In this file add an empty class definition for an indicator. This is a general class that will be derived for every data-processing indicator we use like moving averages, MACD, etc.

class Indicator:
    def __init__(self):
        pass

Order Manager

In the order_management folder, create a file named order_manager.py . In this file, add an empty class definition for our order manager. We will use this to store the trades in a file. I’m making this in a way that you can actually execute the orders too.

class OrderManager:
    def __init__(self):
        pass

Trading Strategy

Same deal. In the trading_strategies folder, add a file named strategy.py . This will be our general strategy class that will be derived in all other trading strategies.

class Strategy:
    def __init__(self):
        pass

Visualizer

In the visualization folder, add a file named visualizer.py . We will use this class to visualize the graph of the price and the indicators in the strategies we’re testing.

Some utility functions before moving on

Utility functions include any function not for the main task, but to help you in achieving extraneous tasks. In that sense, I’d like to have some utility functions to store information in files. Since we’re backtesting trading algorithms, we’re not gonna execute trades, rather we’re gonna store the trades in a file and later read them to find out the net profit/loss.

Create another __init__.py file in the utility folder. After that create a file named utils.py . Let’s first load in the libraries we need in this file:

from datetime import datetime
import os

In this file, we’re gonna make 3 functions:

• The first one will check if a folder exists. This is to ensure that we can create/edit a file in that folder. I’m going to name folders based on strategy and the frequency of trades (seconds, minutes, hours, etc).

def ensure_directory_exists(file_path: str):
    """
    Ensures that the directory for the given file path exists, creating it if necessary.

Args:
        file_path (str): The full path including the filename where the directory existence needs to be checked.
    """
    directory_path = os.path.dirname(file_path)
    if not os.path.exists(directory_path):
        os.makedirs(directory_path)

The function gets the file path, then if the directory doesn’t exist, it creates the directory.

• Getting the right file name. I am going to name files based on current date.

def get_current_date_log_filename(log_directory: str) -> str:
    """Generates a log filename for the current date in the specified directory."""
    return f'{log_directory}/{datetime.now().strftime("%Y-%m-%d")}.txt'

This function gets the folder names (directory), then gets today’s date based on the datetime library and converts it into year-month-day format.

• Formatting time frequency: The frequency at which market data comes to us will be in seconds. To make it easier to read, if we’re looking at minute-by-minute data, its better to say 1 minute rather than 60 seconds. So we can convert the time using this function.

def format_time_duration(seconds: int) -> str:
    """Formats seconds into a human-readable duration string."""
    if seconds % 3600 == 0:
        return "1_hour"
    elif seconds % 60 == 0:
        return "1_min"
    elif seconds % 86400 == 0:
        return "1_day"
    else:
        return f"{seconds}_sec"

If we’re doing 3600 seconds, then its better to say 1 hour. If we’re doing 60 seconds, then its better to say 1 minute . If we’re doing 86400 seconds, then its better to say 1 day. Otherwise, we just return the number of seconds.

Functions to store our trades in files

In the utility folder, add a file named logger.py . This is what we will use to log all of the buy and sell orders in a file.

Importing libraries:

from functools import wraps
from typing import Callable, Any
from utility.utils import (
    get_current_date_log_filename,
    format_time_duration,
    ensure_directory_exists,
)

We’re using functools and typing libraries to annotate our functions so that its easier for future developers to know how to use these functions.

We’re going to write more function wrappers to our buy and sell orders to the order manager. Wrappers act like functions that are run before another function is run. They get the same arguments as the original function. All we need to do is write a wrapper that takes in the price and quantity information from our buy and sell order functions and log them into a file.

Create a function named log_action . This function will return another wrapper function which will return the original buy/sell function. Yes, decorators are just a fancy word for function nesting. This function will look like this:

def log_action(action_type: str, func: Callable) -> Callable:
  @wraps(func)
    def wrapper(self: Any, *args: Any, **kwargs: Any) -> Any:
      ...
      return func(self, *args, **kwargs)
  return wrapper

To explain whats happening. log_action is the name of the function. action_type is something we’ll input. it will either be “buy” or “sell”. It is a string input hence we add the action_type: str type hint. Similarly, it takes another function named func , which is why we add the type hint of Callable to it. Finally, the -> Callable means that the function will return another function.

In the function, we define a wrapper function using the @wraps(func) decorator in Python. It will take in all of the arguments through the args variable and the key-word arguments through the kwargs variable. We will only be using the kwargs variable.

Finally, this wrapper will actually run the function using the func(self, *args, **kwargs) command.

Again, our log_action function merely returns this wrapper function, it doesn’t actually run it.

Getting all the inputs.

Python stores keyword arguments into what we call kwargs . We can just use the kwargs.get() function to get the relevant information in our wrapper function.

price = kwargs.get("price")
time = kwargs.get("time")
quantity = kwargs.get("quantity")
interval = format_time_duration(int(kwargs.get("interval")))
strategy = kwargs.get("strategy")

Using this information, we get our filename

log_filename = get_current_date_log_filename(
    f"logs/orders/{strategy}/{interval}"
)
ensure_directory_exists(log_filename)

Our folder structure is as follows:

• At first we will have a logs folder
• In the logs folder we have another folder called orders
• In this folder, we’ll have different folders for the strategies we’re using
• Finally, each strategy can be run on different frequency. So we’ll have folders for the different intervals we’re running the function on.

Lastly, we ensure that the directory exists before logging to the file.

Brokerage fee

I’m just gonna add two variables for maker and taker brokerage. Maker brokerage is the fees on selling the asset and taker brokerage is the fees on buying the asset.

taker_brokerage: float = 0.0035
maker_brokerage: float = 0.0075

Next we have a condition to check whether the action_type is buy or sell. For each action type, we have a different code.

if action_type == "buy":

  log_message = f"{time} Buy {quantity} @ {price} "
  
  with open(log_filename, "a") as f:
      f.write(log_message)

If the action is buy, then we will simply open the file and add the message “{time} Buy {quantity} @ {price}”

if the action type is sell, then we need to do a few things:

• Get the price when we bought the quantity.
• Calculate the difference in price to get profits,
• Calculate the brokerage fee for the buying and selling
• Log all of that into the file

elif action_type == "sell":

    last_bought = 0

    with open(log_filename, "r") as f:
        last_bought = float(f.read().split("\n")[-1].split(" ")[4])

    profit = round(float(price) * quantity - last_bought * quantity, 3)
    log_message = f" {time} Sell {quantity} @ {price} Profit = {profit} "

    maker_brokerage_fee = round(price * quantity * maker_brokerage, 2)
    taker_brokerage_fee = round(last_bought * quantity * taker_brokerage, 2)

    with open(log_filename, "a") as f:
        f.write(
            log_message + f"Brokerage Fee: {maker_brokerage_fee + taker_brokerage_fee}\n"

We open the file again to get the last line. In the last line, we find the location where we stored the price at which we bought the asset. Then, we convert it into a float.

Then we calculate the profit and the message we want to append.

Finally we calculate the brokerage fee and add that to the file.

from functools import wraps
from typing import Callable, Any
from utility.utils import (
    get_current_date_log_filename,
    format_time_duration,
    ensure_directory_exists,
)

def log_action(action_type: str, func: Callable) -> Callable:
    """Decorator to log buy or sell actions and calculate brokerage fees."""

    @wraps(func)
    def wrapper(self: Any, *args: Any, **kwargs: Any) -> Any:

        price = kwargs.get("price")
        time = kwargs.get("time")
        quantity = kwargs.get("quantity")
        interval = format_time_duration(int(kwargs.get("interval")))
        strategy = kwargs.get("strategy")

        log_filename = get_current_date_log_filename(
            f"logs/orders/{strategy}/{interval}"
        )
        ensure_directory_exists(log_filename)

        taker_brokerage: float = 0.0035
        maker_brokerage: float = 0.0075
        brokerage_fee: float = 0.0

        if action_type == "buy":

            log_message = f"{time} Buy {quantity} @ {price} "

            with open(log_filename, "a") as f:
                f.write(log_message)

        elif action_type == "sell":

            last_bought = 0

            with open(log_filename, "r") as f:
                last_bought = float(f.read().split("\n")[-1].split(" ")[4])

            profit = round(float(price) * quantity - last_bought * quantity, 3)
            log_message = f" {time} Sell {quantity} @ {price} Profit = {profit} "
            maker_brokerage_fee = round(price * quantity * maker_brokerage, 2)
            taker_brokerage_fee = round(last_bought * quantity * taker_brokerage, 2)
        
            with open(log_filename, "a") as f:
                f.write(
                    log_message + f"Brokerage Fee: {maker_brokerage_fee + taker_brokerage_fee}\n"
                )

        return func(self, *args, **kwargs)

    return wrapper

Now that the hard work is over, we can simply have two functions which run this function, one for buy and one for sell.

In the same file add two more functions:

def log_buy_order(func: Callable) -> Callable:
    """Decorator to log buy orders."""
    return log_action("buy", func)

def log_sell_order(func: Callable) -> Callable:
    """Decorator to log sell orders."""
    return log_action("sell", func)

We have all of the utility and logging functions set up. We’re finally ready to code the main functions of the project. Stay tuned for the next article.